package SparkRDD.RDD的缓存

import org.apache.commons.lang.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

class test {


  @Test
  def test: Unit ={
    // 创建sc
    val conf = new SparkConf().setAppName("ip统计").setMaster("local[6]")
    val sc   = new SparkContext(conf)

    // 读取数据
    val resource = sc.textFile("src/main/scala/RDD的缓存/ip.txt")

    // 取出ip，赋予词频
    val ipRDD = resource.map( item => ( item.split(",")(0) , 1) )

    // 简单的数据清洗
    val cleanRDD = ipRDD.filter( item => StringUtils.isNotEmpty(item._1) )

    // 统计ip出现的次数
    val aggRDD = cleanRDD.reduceByKey( (curr,agg) => curr + agg )

    // 提取出ip统计出现最多的
    val maxRDD = aggRDD.sortBy( item => item._2,ascending = true ).first()   // Action操作1
    // 提取出ip统计出现最少的
    val minRDD = aggRDD.sortBy( item => item._2,ascending = false ).first()  // Action操作2

    println("max:"+maxRDD,"min:"+minRDD)

    /**
     * ResultStage 9 (first at test.scala:31) finished in 0.006 s
     * Job 3 finished: first at test.scala:31, took 0.048808 s
     */

    /**
     *  在这里我们可以看到求取最大iP次数与最小ip次数，都是在aggRDD基础之上进行的，
     *  并且每进行一次Action操作，都会执行一次Transformation操作
     *  而一次执行就会进行两次shuffle操作（reduceByKey、sortBy）
     *  总共就进行了 4次 shuffle操作，如果数据量十分庞大，每次Action都会重新运算，显得十分繁琐
     *  优化方案：
     *  在进行Action操作之前，我们可以将器缓存下来，下次需要时直接调用即可
     *
     */


  }

}
